package com.wj.core.common.util;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author: wangjun
 * @create: 2018-06-04 
 **/
public class ZkUtil {

  private final static Logger logger = LoggerFactory.getLogger(ZkUtil.class);

  private static ZooKeeper zooKeeper;
  private static ReentrantLock lock = new ReentrantLock(true);

  public static ZooKeeper getInstance() {
    if (zooKeeper == null) {

      try {
        if (lock.tryLock(2, TimeUnit.SECONDS)) {
          zooKeeper = new ZooKeeper(EnvironmentUtil.ZK_ADDRESS, 30000, new Watcher() {
            @Override
            public void process(final WatchedEvent event) {
              //会话过期
              if (event.getState() == Event.KeeperState.Expired) {
                try {
                  zooKeeper.close();
                } catch (InterruptedException e) {
                  logger.info("zookeeper close fail :{}", e);
                }
                zooKeeper = null;
              }

              try {
                String znodePath = event.getPath();
                if (znodePath != null) {
                  zooKeeper.exists(znodePath, true);
                }
              } catch (KeeperException e) {
                logger.info("zookpeer exception :{}", e);
              } catch (InterruptedException e) {
                logger.info("zookpeer exception :{}", e);
              }

              //刷新service address
              if (event.getType() == Event.EventType.NodeChildrenChanged
                  || event.getState() == Event.KeeperState.SyncConnected) {
                freshServiceAddress();
              }
            }
          });
        }
      } catch (InterruptedException e) {
        logger.error("connect fail ：{}", e);
      } catch (IOException e) {
        logger.error("connect fail ：{}", e);
      }
    }
    return zooKeeper;
  }

  public static void freshServiceAddress() {
    ConcurrentMap<String, Set<String>> tempMap = new ConcurrentHashMap<String, Set<String>>();
    try {
      // iface list
      List<String> interfaceNameList = getInstance().getChildren(EnvironmentUtil.ZK_SERVICE_PATH, true);

      if (interfaceNameList != null && interfaceNameList.size() > 0) {
        for (String interfaceName : interfaceNameList) {

          // address list
          String ifacePath = EnvironmentUtil.ZK_SERVICE_PATH.concat("/").concat(interfaceName);
          List<String> addressList = getInstance().getChildren(ifacePath, true);

          if (addressList != null && addressList.size() > 0) {
            Set<String> addressSet = new HashSet<String>();
            for (String address : addressList) {

              // data from address
              String addressPath = ifacePath.concat("/").concat(address);
              byte[] bytes = getInstance().getData(addressPath, false, null);
              addressSet.add(new String(bytes));
            }
            tempMap.put(interfaceName, addressSet);
          }
        }
        serviceAddress = tempMap;
        logger.info(">>>>>>>>>>> wj-rpc fresh serviceAddress success: {}", serviceAddress);
      }
    } catch (KeeperException e) {
      logger.error("", e);
    } catch (InterruptedException e) {
      logger.error("", e);
    }
  }

  public static volatile ConcurrentMap<String, Set<String>> serviceAddress = new ConcurrentHashMap<>();
}
